package com.lfd.rocketmq;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lfd.rocketmq.pojo.Order;
import com.lfd.rocketmq.service.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.io.IOException;
import java.util.*;

@Slf4j
@SpringBootApplication
public class RocketMQApplication implements ApplicationRunner {
    @Autowired
    private ConsumerProducerRockMQ consumerProducerRockMQ;
    @Autowired
    private ProducerProducerRockMQ producerProducerRockMQ;
    @Autowired
    private AsyncProducerRockMQ asyncProducerRockMQ;
    @Autowired
    private OnewayProducerRocketMQ onewayProducerRocketMq;
    @Autowired
    private DelayProducerRockMQ delayProducerRockMQ;
    @Autowired
    private ObjectMapper objectMapper;

    String consumerGroup = "example_consumer_group";
    String namesrvAddr = "localhost:9876";
    String tag = "example_tag";
    String topic = "example_topic";

    @Override
    public void run(ApplicationArguments args) throws Exception {
//        testProducerRockMQ();
//        testAsyncProducerRockMQ();
//        testOnewayRockMQ();
//        testDelayProducerRockMQ();
//        testOrderly();
//        testBatch();
//        testTransaction();
//        testTag();
//        testKey();
//        testRetry();
//        testDeadMsg();
        testRepeat();
    }


    /**
     * 测试同步send
     *
     * @param
     */
    void testProducerRockMQ() throws MQClientException, MQBrokerException, RemotingException, InterruptedException, JsonProcessingException {
        //        4. 运行示例
//        启动 RocketMQ NameServer 和 Broker。
//        首先运行消费者代码，它会订阅消息。
        consumerProducerRockMQ.registerMessageListener(consumerGroup, namesrvAddr, topic, tag);
//        然后运行生产者代码，它会发送消息到主题。
//        消费者会接收到这些消息并打印到控制台。
        producerProducerRockMQ.send(namesrvAddr, topic, tag, "hello world");
    }

    /**
     * 测试异步send
     *
     * @param
     */

    void testAsyncProducerRockMQ() throws MQClientException, MQBrokerException, RemotingException, InterruptedException, IOException {
        //        4. 运行示例
//        启动 RocketMQ NameServer 和 Broker。
//        首先运行消费者代码，它会订阅消息。
        consumerProducerRockMQ.registerMessageListener(consumerGroup, namesrvAddr, topic, tag);
//        然后运行生产者代码，它会发送消息到主题。
//        消费者会接收到这些消息并打印到控制台。
        asyncProducerRockMQ.sendAsyncMsg(namesrvAddr, topic, tag, "hello world zzz ");
    }

    /**
     * 测试单向消息send
     *
     * @param
     */
    void testOnewayRockMQ() throws MQClientException, MQBrokerException, RemotingException, InterruptedException, IOException {
        //        4. 运行示例
//        启动 RocketMQ NameServer 和 Broker。
//        首先运行消费者代码，它会订阅消息。
        consumerProducerRockMQ.registerMessageListener(consumerGroup, namesrvAddr, topic, tag);
//        然后运行生产者代码，它会发送消息到主题。
//        消费者会接收到这些消息并打印到控制台。
        onewayProducerRocketMq.sendOnewayMsg(namesrvAddr, topic, tag, "hello world 单向消息 ");
    }

    /**
     * 13.RocketMQ发送延迟消息
     *
     * @param
     */
    void testDelayProducerRockMQ() throws MQClientException, MQBrokerException, RemotingException, InterruptedException, IOException {
        //        4. 运行示例
//        启动 RocketMQ NameServer 和 Broker。
//        首先运行消费者代码，它会订阅消息。
        consumerProducerRockMQ.registerMessageListener(consumerGroup, namesrvAddr, topic, tag);
//        然后运行生产者代码，它会发送消息到主题。
//        消费者会接收到这些消息并打印到控制台。
        delayProducerRockMQ.send(namesrvAddr, topic, tag, "hello world发送延迟消息");

    }

    private void testOrderly() {
        try {
            testOrderlyProducer();
            testOrderlyConsumer();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 14.3顺序消息生产者
     */
    public void testOrderlyProducer() throws MQClientException, MQBrokerException, RemotingException, InterruptedException, IOException {
        // 创建默认的生产者
        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        // 设置nameServer地址
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        List<Order> orders = Arrays.asList(new Order(1, 111, 59D, new Date(), "下订单"), new Order(2, 111, 59D, new Date(), "物流"), new Order(3, 111, 59D, new Date(), "签收"), new Order(4, 112, 89D, new Date(), "下订单"), new Order(5, 112, 89D, new Date(), "物流"), new Order(6, 112, 89D, new Date(), "拒收"));
        // 循环集合开始发送
        orders.stream().forEach(order -> {
            try {
                String json = objectMapper.writeValueAsString(order);
                Message message = new Message("TopicTest", json.getBytes());
                // 发送的时候 相同的订单号选择同一个队列
                producer.send(message, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        // 当前主题有多少个队列
                        int queueNumber = mqs.size();
                        // 这个arg就是后面传入的 order.getOrderNumber()
                        Integer i = (Integer) arg;
                        // 用这个值去%队列的个数得到一个队列
                        int index = i % queueNumber;
                        // 返回选择的这个队列即可 ，那么相同的订单号 就会被放在相同的队列里 实现FIFO了
                        try {
                            log.info("顺序消息生产者 msg :{}", objectMapper.writeValueAsString(mqs.get(index)));
                        } catch (JsonProcessingException e) {
                            throw new RuntimeException(e);
                        }
                        return mqs.get(index);
                    }
                }, order.getOrderNumber());
            } catch (JsonProcessingException | MQClientException | RemotingException | MQBrokerException |
                     InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 14.4顺序消息消费者，测试时等一会即可有延迟
     */
    public void testOrderlyConsumer() throws Exception {
        // 创建默认消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        // 设置nameServer地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅一个主题来消费   *表示没有过滤参数 表示这个主题的任何消息
        consumer.subscribe("TopicTest", "*");
        // 注册一个消费监听 MessageListenerOrderly 是顺序消费 单线程消费
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                MessageExt messageExt = msgs.get(0);
                log.info("顺序消息消费者 messageExt :{}", new String(messageExt.getBody()));
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();

    }

    public static void main(String[] args) {
        SpringApplication.run(RocketMQApplication.class, args);
    }

    private void testBatch() {
        try {
            testBatchProducer();
            testBatchConsumer();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 15.1批量消息生产者
     */
    public void testBatchProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        List<Message> list = Arrays.asList(new Message("TopicTest", "我是一组消息的A消息".getBytes()), new Message("TopicTest", "我是一组消息的B消息".getBytes()), new Message("TopicTest", "我是一组消息的C消息".getBytes())

        );
        producer.send(list);

    }

    /**
     * 15.2批量消息消费者
     */
    public void testBatchConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "*");
        //注册一个消费监听 MessageListenerConcurrently是并发消费
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 这里执行消费的代码 默认是多线程消费
                log.info("批量消息消费者:{},msg数据{}", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    private void testTransaction() {
        try {
            testTransactionProducer();
            testTransactionConsumer();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 16.2事务消息生产者
     * TransactionalMessageCheckService的检测频率默认1分钟，可通过在broker.conf文件中设置transactionCheckInterval的值来改变默认值，单位为毫秒。
     * 从broker配置文件中获取transactionTimeOut参数值。
     * 从broker配置文件中获取transactionCheckMax参数值，表示事务的最大检测次数，如果超过检测次数，消息会默认为丢弃，即回滚消息。
     *
     * @throws Exception
     */
    public void testTransactionProducer() throws Exception {
        // 创建一个事务消息生产者
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("test-group");
        transactionMQProducer.setNamesrvAddr("localhost:9876");
        // 设置事务消息监听器
        transactionMQProducer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // 这个是执行本地业务方法
                log.info("事务消息生产者 executeLocalTransaction 时间:{} msg:{}", new Date(), new String(msg.getBody()));
                // 这个可以使用try catch对业务代码进行性包裹
                // COMMIT_MESSAGE 表示允许消费者消费该消息
                // ROLLBACK_MESSAGE 表示该消息将被删除，不允许消费
                // UNKNOW表示需要MQ回查才能确定状态 那么过一会 代码会走下面的checkLocalTransaction(msg)方法

                // return LocalTransactionState.UNKNOW;
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // 这里是回查方法 回查不是再次执行业务操作，而是确认上面的操作是否有结果
                // 默认是1min回查 默认回查15次 超过次数则丢弃打印日志 可以通过参数设置
                // transactionTimeOut 超时时间
                // transactionCheckMax 最大回查次数
                // transactionCheckInterval 回查间隔时间单位毫秒
                // 触发条件
                // 1.当上面执行本地事务返回结果UNKNOW时,或者下面的回查方法也返回UNKNOW时 会触发回查
                // 2.当上面操作超过20s没有做出一个结果，也就是超时或者卡主了，也会进行回查
                log.info("事务消息生产者 checkLocalTransaction 时间:{} msg:{}", new Date(), new String(msg.getBody()));
                // return LocalTransactionState.UNKNOW;
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        transactionMQProducer.start();
        Message message = new Message("TopicTest2", "我是一个事务消息".getBytes());
        transactionMQProducer.sendMessageInTransaction(message, null);
        log.info("sendMessageInTransaction 时间:{}", new Date());
    }

    /***
     * 16.3事务消息消费者
     */
    public void testTransactionConsumer() throws Exception {
        // 创建默认消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        // 设置nameServer地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅一个主题来消费   *表示没有过滤参数 表示这个主题的任何消息
        consumer.subscribe("TopicTest2", "*");
        // 注册一个消费监听 MessageListenerConcurrently是并发消费
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 默认是20个线程一起消费，可以参看 consumer.setConsumeThreadMax()
                // 这里执行消费的代码 默认是多线程消费
                log.info("事务消息消费者 线程名:{},消息内容:{}", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    private void testTag() {
        try {
            testTagProducer();
            testTagConsumer();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 17.RocketMQ发送带标签的消息，消息过滤
     * Rocketmq提供消息过滤功能，通过tag或者key进行区分
     * 我们往一个主题里面发送消息的时候，根据业务逻辑，可能需要区分，比如带有tagA标签的被A消费，
     * 带有tagB标签的被B消费，还有在事务监听的类里面，只要是事务消息都要走同一个监听，我们也需要通过过滤才区别对待
     * 17.1标签消息生产者
     */
    public void testTagProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        Message message = new Message("TopicTest", "tagA", "我是一个带标记的消息".getBytes());
        SendResult send = producer.send(message);
        log.info("标签消息生产者send:{}", objectMapper.writeValueAsString(send));
    }

    /**
     * 17.2标签消息消费者
     *
     * @throws Exception
     */
    public void testTagConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅一个主题来消费   表达式，默认是*,支持"tagA || tagB || tagC" 这样或者的写法 只要是符合任何一个标签都可以消费
        consumer.subscribe("TopicTest", "tagA || tagB || tagC");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 这里执行消费的代码 默认是多线程消费
                log.info("标签消息消费者 name:{},消息体:{},topic:{}", Thread.currentThread().getName(), new String(msgs.get(0).getBody()), msgs.get(0).getTopic());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    private void testKey() {
        try {
            testKeyProducer();
            testKeyConsumer();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 18.1带key消息生产者
     */
    private void testKeyProducer() throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("test-group");
        defaultMQProducer.setNamesrvAddr("localhost:9876");
        defaultMQProducer.start();
        defaultMQProducer.send(new Message("TopicTest", "tagA", "key", "我是一个带标记和key的消息".getBytes()));
        log.info("带key消息生产者:{}", new Date());
    }

    /***
     * 18.2带key消息消费者
     */
    public void testKeyConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "tagA || tagB || tagC");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                log.info("带key消息消费者 线程名:{},msg:{},key:{}", Thread.currentThread().getName(), new String(msgs.get(0).getBody()), msgs.get(0).getKeys());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    /**
     * 19.RocketMQ重试机制
     * 19.1生产者重试
     * // 失败的情况重发3次
     * producer.setRetryTimesWhenSendFailed(3);
     * // 消息在1S内没有发送成功，就会重试
     * producer.send(msg, 1000);
     * 19.2消费者重试
     * 在消费者放return ConsumeConcurrentlyStatus.RECONSUME_LATER;后就会执行重试
     * 上图代码中说明了，我们再实际生产过程中，一般重试3-5次，如果还没有消费成功，则可以把消息签收了，通知人工等处理
     */

    private void testRetry() {
        try {
            testRetryProducer();
            testRetryConsumer();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void testRetryProducer() throws MQBrokerException, RemotingException, InterruptedException, MQClientException, JsonProcessingException {
        DefaultMQProducer producer = new DefaultMQProducer("retry-producer-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.setRetryTimesWhenSendFailed(3);
        producer.start();
        for (int i = 0; i < 3; i++) {

            try {
                Message message;
                // 模拟发送消息失败，通过抛异常来触发重试
                if (i == 2) {
//                    throw new RuntimeException("模拟发送失败");
                    message = null;
                } else {
                    message = new Message("TopicTest", "body".getBytes());
                }
                SendResult send = producer.send(message);
                log.info("testRetryProducer:{}", objectMapper.writeValueAsString(send));
            } catch (Exception e) {
//                e.printStackTrace();
//                log.error("Exception:{}", e);
            }
        }
        log.info("testRetryProducer");
        // 关闭生产者
//        producer.shutdown();
    }


    /***
     * 19.2带key消息消费者 2消费者重试
     */
    public void testRetryConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                log.info("带key2消息消费者 线程名:{},msg:{},key:{}", Thread.currentThread().getName(), new String(msgs.get(0).getBody()), msgs.get(0).getKeys());
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        consumer.start();
    }

    private void testDeadMsg() {
        try {
            testDeadMsgProducer();
            testDeadMsgConsumer();
            // 等待一段时间，让消息进入死信队列
            Thread.sleep(5000);
            testDeadMq();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 20.RocketMQ死信消息
     * 当消费重试到达阈值以后，消息不会被投递给消费者了，而是进入了死信队列
     * 当一条消息初次消费失败，RocketMQ会自动进行消息重试，达到最大重试次数后，若消费依然失败，
     * 则表明消费者在正常情况下无法正确地消费该消息。此时，该消息不会立刻被丢弃，而是将其发送到该消费者对应的特殊队列中，
     * 这类消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue），
     * 死信队列是死信Topic下分区数唯一的单独队列。如果产生了死信消息，那对应的ConsumerGroup的死信Topic名称为%DLQ%ConsumerGroupName，
     * 死信队列的消息将不会再被消费。可以利用RocketMQ Admin工具或者RocketMQ Dashboard上查询到对应死信消息的信息。
     * 我们也可以去监听死信队列，然后进行自己的业务上的逻辑
     * 20.1消息生产者
     */
    public void testDeadMsgProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("dead-msg-producer-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        // 失败的情况重发3次
        producer.setRetryTimesWhenSendFailed(3);
        Message message = new Message("TopicTest", "我是一个死信消息".getBytes());
        producer.send(message);
//        producer.shutdown();
        log.info("消息生产者 testDeadMsgProducer:{}", message);
    }

    /**
     * 20.2消息消费者
     */
    public void testDeadMsgConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
                "dead-msg-producer-group");
        consumer.setNamesrvAddr("localhost:9876");
        //设置最大消费重试次数 2 次
        consumer.setMaxReconsumeTimes(2);
        consumer.subscribe("TopicTest", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                log.info("消息消费者 consumeMessage:{}", new String(msgs.get(0).getBody())); // 打印当前重试次数
                log.info("消息重试次数: {}", msgs.get(0).getReconsumeTimes());
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        consumer.start();
    }

    /**
     * 20.3死信消费者
     */
    public void testDeadMq() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
                "dead-msg-producer-group");
        consumer.setNamesrvAddr("localhost:9876");
        // 消费重试到达阈值以后，消息不会被投递给消费者了，而是进入了死信队列
        // 队列名称 默认是 %DLQ% + 消费者组名
        consumer.subscribe("%DLQ%dead-msg-producer-group", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                log.info("死信消费者 consumeMessage:{}", new String(msgs.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    private void testRepeat() {
        try {
            testRepeatProducer();
            testRepeatConsumer();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 21.3测试生产者
     *
     * @throws Exception
     */
    public void testRepeatProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("repeat-producer-group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        String keyId = UUID.randomUUID().toString();
        log.info("repeatProducer keyId:{}", keyId);
        Message message = new Message("TopicTest", "tagA", keyId, "我是一个测试消息".getBytes());
        producer.send(message);
    }

    /**
     * 在boot项目中可以使用@Bean在整个容器中放置一个单利对象
     */
    public static BitMapBloomFilter bloomFilter = new BitMapBloomFilter(100);

    /**
     * 21.5测试消费者
     */
    public void testRepeatConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("repeat-producer-group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TopicTest", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                log.info("测试消费者Repeat:{},uuid:{}", new String(msgs.get(0).getBody()), new String(msgs.get(0).getKeys()));
                MessageExt messageExt = msgs.get(0);
                String keys = messageExt.getKeys();
                if (bloomFilter.contains(keys)) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                boolean add = bloomFilter.add(keys);
                log.info("测试消费者Repeat add:{}", add);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}
